package com.swak.reactivex.cluster.impl;

import static com.swak.Constants.CLIENT_SIDE;
import static com.swak.Constants.SIDE;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.exception.RemotingException;
import com.swak.reactivex.cluster.Cluster;
import com.swak.reactivex.cluster.ClusterExchangeHandler;
import com.swak.reactivex.cluster.ID;
import com.swak.reactivex.cluster.Member;
import com.swak.reactivex.cluster.Members;
import com.swak.reactivex.transport.Channel;
import com.swak.reactivex.transport.ExchangeChannel;
import com.swak.reactivex.transport.ExchangeHandler;
import com.swak.reactivex.transport.Message;

/**
 * ClusterHandler.
 * 
 * @author DELL
 */
public class ClusterHandler implements ClusterExchangeHandler {

	private static final Logger LOGGER = LoggerFactory.getLogger(Cluster.class);
	private static final String RemoteID = "RemoteID";

	private final Cluster cluster;
	private final ExchangeHandler delegate;

	public ClusterHandler(Cluster cluster, ExchangeHandler delegate) {
		this.cluster = cluster;
		this.delegate = delegate;
	}

	@Override
	public void memberShipChanaged(Member member, List<Member> members, boolean add) {
		if (delegate instanceof ClusterExchangeHandler) {
			((ClusterExchangeHandler) delegate).memberShipChanaged(member, members, add);
		}
	}

	@Override
	public void connected(Channel channel) throws RemotingException {
		String side = (String) channel.getAttribute(SIDE);
		if (CLIENT_SIDE.equals(side)) {

			// 仅仅保存客户端连接， 服务端连接等收到 交换ID请求之后才保存
			ID id = ID.remoteID(channel.url());
			this.register(id, (ExchangeChannel) channel);

			this.sendLocalID(channel);
			if (this.delegate != null) {
				this.delegate.connected(channel);
			}
		}
	}

	private void sendLocalID(Channel channel) throws RemotingException {
		channel.send(Message.forMessage().setData(this.cluster.member().id()));
		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("[Cluster Building] Client Send: " + this.cluster.member().id());
		}
	}

	private void recvRemoteID(Channel channel, ID peerID) throws RemotingException {
		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("[Cluster Building] Server Recv: " + peerID);
		}

		channel.setAttribute(RemoteID, peerID);
		this.register(peerID, (ExchangeChannel) channel);
		if (this.delegate != null) {
			this.delegate.connected(channel);
		}
	}

	@Override
	public void disconnected(Channel channel) throws RemotingException {

		/*
		 * 关闭本端的连接
		 */
		String side = (String) channel.getAttribute(SIDE);
		if (CLIENT_SIDE.equals(side)) {
			ID id = ID.remoteID(channel.url());
			this.unRegister(id, (ExchangeChannel) channel);
		}

		/*
		 * 关闭远端的连接
		 */
		else {
			ID id = (ID) channel.getAttribute(RemoteID);
			if (id != null) {
				this.unRegister(id, (ExchangeChannel) channel);
			}
		}

		/*
		 * 传递关闭连接事件
		 */
		if (this.delegate != null) {
			this.delegate.disconnected(channel);
		}
	}

	private void register(ID id, ExchangeChannel channel) {
		Members members = this.cluster.members();
		Member member = members.check(id);
		if (member == null) {
			synchronized (members) {
				member = members.check(id);
				if (member == null) {
					member = new MemberImpl(this.cluster, id);
					members.register(member);
				}
			}
		}
		members.register(member, channel);
	}

	private void unRegister(ID id, ExchangeChannel channel) {
		Members members = this.cluster.members();
		Member member = members.check(id);
		if (member != null) {
			member.channels().remove((ExchangeChannel) channel);
			if (member.channels().isEmpty()) {
				members.unregister(member);
			}
		}
	}

	@Override
	public void sent(Channel channel, Object message) throws RemotingException {
		if (this.delegate != null) {
			this.delegate.sent(channel, message);
		}
	}

	@Override
	public void received(Channel channel, Object message) throws RemotingException {
		Object msg = ((Message) message).getData();
		if (msg instanceof ID) {
			this.recvRemoteID(channel, (ID) msg);
			return;
		}
		if (this.delegate != null) {
			this.delegate.received(channel, message);
		}
	}

	@Override
	public CompletionStage<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
		if (this.delegate != null) {
			return this.delegate.reply(channel, message);
		}
		return CompletableFuture.completedStage(null);
	}

	@Override
	public void caught(Channel channel, Throwable exception) throws RemotingException {
		if (this.delegate != null) {
			this.delegate.caught(channel, exception);
		}
	}
}
